#ifndef HASHJOINBOX_H_JM_20120910
#define HASHJOINBOX_H_JM_20120910

#include "HashActivityBox.h"


namespace jm4bbx {

	//struct HashJoinBoxDescriptor {
	//	static const  groupTypeINPUT_GROUP;
	//	static const  groupTypeOUTPUT_GROUP;	
	//	static const  columnType INPUT_COLUMN;
	//	static const  columnType OUTPUT_COLUMN;
	//	static const  inType IN_ARC;
	//	static const  bobox::outarc_index_type OUT_ARC;

	//};

	template</*class HashBoxDescriptorType,*/ class HashBoxType>
	class HashJoinBox : 
		public HashActivityBox</*HashBoxDescriptorType,*/ HashBoxType> {
	public:
		HashJoinBox(bobox::box_id_type id, bobox::box_tid_type tid, 
			bobox::request *req, const bobox::parameters &params) 
			: HashActivityBox(id, tid, req, params)	{ }
	protected:
		
		virtual bool synchronize() const  override { return false; };

		virtual void handleRecord(valueType value, ExtendedHashEntry& entry) override {
			std::stringstream stream;
			for (size_t i = 0; i < entry.second; ++i) {
				stream << "join_box id:" << getBoxInstanceId() << "  vykona ("
					<<  entry.first.first << "," << entry.first.second 
					<< ") join " << value << "\n";
				std::cout << stream.str();
				stream.str(std::string());
				stream.clear();
			}	
		} 

	};

	//template<class BoxDescriptor, class hashBoxType>
	//class  HashJoinBox : public Box<BoxDescriptor> {
	//	
	//public:

	//	typedef typename hashBoxType::StaticHashTableHolderType StaticHashTableHolderType;
	//	typedef typename StaticHashTableHolderType::HashTableType::HashEntry HashEntry;
	//	typedef typename StaticHashTableHolderType::HashTableType::ExtendedHashEntry ExtendedHashEntry;

	//	typedef typename hashBoxType::policyType policyType;
	//	typedef typename hashBoxType::valueType valueType;

	//	HashJoinBox(bobox::box_id_type id, bobox::box_tid_type tid, bobox::request *req, 
	//		const bobox::parameters &params) 
	//		: Box(id, tid, req, params)	{ }

	//	virtual void init_impl() {

	//		add_group(BoxIO::INPUT_GROUP, 
	//			BoxIO::INPUT_COLUMN, 
	//			get_input_descriptors(BoxIO::IN_ARC));

	//		add_group(BoxIO::OUTPUT_GROUP, 
	//			BoxIO::INPUT_COLUMN, 
	//			get_output_descriptors(BoxIO::OUT_ARC));

	//		wait4Receive();
	//	}

	//	virtual void sync_mach_etwas() {

	//		const group& ig = get_group(BoxIO::INPUT_GROUP);
	//		if (ig.poisoned_) {
	//			send_poisoned(BoxIO::OUT_ARC);
	//			return;
	//		}

	//		const int rpos = ig.rdpos_;
	//		const int rend = ig.rdend_;

	//		BoostHashWrapper<valueType> hash;


	//		tableHolder_.getHashTable()->setSynchronized(false);
	//		
	//		auto bid = boxInstanceId_;
	//		int counter = 0;
	//		
	//		for (; ig.rdpos_ < ig.rdend_; ++ig.rdpos_) {
	//			
	//			auto key = get_data<policyType>(BoxIO::INPUT_COLUMN)[ig.rdpos_];
	//			
	//			tableHolder_.getHashTable()->foreachExtended(key, [bid, key, &counter](const ExtendedHashEntry& entry) {
	//				std::stringstream stream;
	//				for (size_t i = 0; i < entry.second; ++i) {
	//					stream << "join_box id:" << bid << "  vykona ("
	//						<<  entry.first.first << "," << entry.first.second 
	//						<< ") join " << key << "\n";
	//					std::cout << stream.str();
	//					stream.clear();
	//				}					
 //				});
	//		
	//				
	//		}

	//		const group& og = get_group(BoxIO::OUTPUT_GROUP);

	//		og.wrbegin_ = rpos;
	//		og.wrpos_ = rend;
	//		og.wrend_ = rend;

	//		send(BoxIO::OUT_ARC, BoxIO::OUTPUT_GROUP);

	//		wait4Receive();

	//	}

	//
	//private:
	//
	//	StaticHashTableHolderType tableHolder_;

	//	void wait4Receive() {
	//		receive(BoxIO::IN_ARC, BoxIO::INPUT_GROUP);
	//	}
	//};

	//struct IntegerHashJoinBoxDescriptor : public HashActivityBoxDescriptor {
	//	//static const char* BOX_ID;
	//};

	//struct StringHashJoinBoxDescriptor : public HashActivityBoxDescriptor {
	//	//static const char* BOX_ID;
	//};

	class IntegerHashJoinBox : 
		public HashJoinBox</*IntegerHashJoinBoxDescriptor,*/ IntegerHashBox> {
	public:
		IntegerHashJoinBox(bobox::box_id_type id, bobox::box_tid_type tid, 
			bobox::request *req, const bobox::parameters &params) 
			: HashJoinBox(id, tid, req, params)	{ }

	};


	class StringHashJoinBox : 
		public HashJoinBox</*StringHashJoinBoxDescriptor,*/ StringHashBox> {
	public:
		StringHashJoinBox(bobox::box_id_type id, bobox::box_tid_type tid, 
			bobox::request *req, const bobox::parameters &params) 
			: HashJoinBox(id, tid, req, params)	{ }

	};

} // namespace jm4bbx


#endif
